package com.hadoop.avg;


import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;

public class AvgReducer extends
Reducer<Text, IntWritable, Text, IntWritable> {
	
	private MultipleOutputs<Text, IntWritable> mos;
	
	
	//reduce 출력키
	private Text outputKey = new Text();
	//reduce 출력값 
	private IntWritable result = new IntWritable();

	@Override
	public void setup(Context context) throws IOException, InterruptedException{
		mos = new MultipleOutputs<Text, IntWritable>(context);
	}
	
	@Override
	public void cleanup(Context context) throws IOException, InterruptedException{
		mos.close();
	}
	
	public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException{
		int sum = 0;
		for(IntWritable value : values)
			sum += value.get();
		
		outputKey.set(key);
		result.set(sum);
	
		
		if(key.toString().indexOf("Avg") >= 0){
			mos.write("Average", outputKey, result);//특정 이름의 다른 파일로 데이터 저장
		}else if(key.toString().indexOf("Add") >= 0) {
			mos.write("Sum", outputKey, result);
		}else{
			context.write(outputKey, result); //기본으로 Part-nnnn으로 저장됨 (파티션별로 저장)
		}
	}
}
